<?php

abstract class QueueConsumer {
	const SIGNAL_RESTART = 11;
	/*
	*	"11" : Linux Signal SIGUSR1,
	*   Linux系统中预留的用户自定义信号，用于通知外部守护进程重启
	*/

	protected $queue_name;

	public function __construct($queue_name = null) {
		$this->queue_name = is_null($queue_name) ? get_class($this) : $queue_name;
	}

  public function consume($data) {
    if (ON_PRODUCTION) return;
    return $this->worker($data);
  }

  public function doWorker() {
    $q = new Queue($this->queue_name);
    $start = time();
    while(time() - $start < (ON_PRODUCTION ? 90 : 1)) { //暂时是60s启动一次，每次90s停止。
      $data = $q->pop();
      if ($data === false) {
        sleep(1);//usleep(50000);
      } else {

        try {
          $this->worker($data);
        } catch (Exception $e) {

          $retry = isset($data['_retry']) ? intval($data['_retry']) : 0;

          if ($retry > 30) {
            throw $e;

          } else {
            $data['_retry'] = $retry + 1;
            $q->push($data);
          }
        }

        if ($this->getMemoryUsage() > 0.9) {
          $this->restart();	//防止内存爆掉
        }
      }
    }
  }

	//每个子类都把自己的处理逻辑放到这里
	abstract protected function worker($message);

	protected function log($log_string) {
    file_put_contents('/tmp/consumer_' . get_class($this), $log_string . "\n", FILE_APPEND);
		//SystemLog::local_log('consumer_' . get_class($this), $log_string);
	}

	protected function restart() {
		exit(self::SIGNAL_RESTART);
	}

	private function getMemoryUsage(){
		$limit = $this->convertStorageUnit(ini_get('memory_limit'));
		$peak = memory_get_peak_usage(true);
		return $peak/$limit;
	}
	/*
	 * @param  string  $size	eg: 10KB | 10K | 10M
	 * @return integer
	 * */
	private function convertStorageUnit($size) {
		$size = rtrim(strtoupper($size), 'B');
		$units = array('K', 'M', 'G', 'T');
		preg_match("/([0-9]+)([KMGT]?)/", $size, $matches);
		if (isset($matches[2])) {
			return $matches[1] * pow(1024, array_search($matches[2], $units) + 1);
		} else {
			return $matches[1];
		}
	}


}
